package codec

import (
	"bufio"
	"hash/crc32"
	"io"
	"myRCP/compressor"
	"myRCP/header"
	"myRCP/serializer"
	"net/rpc"
	"sync"
)

type clientCodec struct {
	r io.Reader
	w io.Writer
	c io.Closer

	compressor compressor.CompressType // rpc 压缩类型 (raw,gzip,snappy,zlib)
	serializer serializer.Serializer   // 表示使用的序列化器
	response   header.ResponseHeader   // rpc 响应的头部
	mutex      sync.Mutex              // 用于保护 pending 的互斥锁
	pending    map[uint64]string
}

//创建一个新的客户端编解码器
//通过缓冲IO来提高RPC的读写性能
func NewClientCodec(conn io.ReadWriteCloser,
	compressType compressor.CompressType, serializer serializer.Serializer) rpc.ClientCodec {

	return &clientCodec{
		r:          bufio.NewReader(conn),
		w:          bufio.NewWriter(conn),
		c:          conn,
		compressor: compressType,
		serializer: serializer,
		pending:    make(map[uint64]string),
	}
}

//写入请求将rpc请求头和主体写入io流
func (c *clientCodec) WriteRequest(r *rpc.Request, param interface{}) error {
	c.mutex.Lock()
	c.pending[r.Seq] = r.ServiceMethod
	c.mutex.Unlock()
	// 判断压缩器是否存在
	if _, ok := compressor.Compressors[c.compressor]; !ok {
		return NotFoundCompressorError
	}
	reqBody, err := c.serializer.Marshal(param) // 用序列化器进行编码
	if err != nil {
		return err
	}
	// 压缩
	compressedReqBody, err := compressor.Compressors[c.compressor].Zip(reqBody)
	if err != nil {
		return err
	}
	// 从请求头部对象池取出请求头
	h := header.RequestPool.Get().(*header.RequestHeader)
	defer func() {
		h.ResetHeader()
		header.RequestPool.Put(h)
	}()
	h.ID = r.Seq
	h.Method = r.ServiceMethod
	h.RequestLen = uint32(len(compressedReqBody))
	h.CompressType = compressor.CompressType(c.compressor)
	h.Checksum = crc32.ChecksumIEEE(compressedReqBody)
	// 发送请求头
	if err := sendFrame(c.w, h.Marshal()); err != nil {
		return err
	}
	// 发送请求体
	if err := write(c.w, compressedReqBody); err != nil {
		return err
	}
	c.w.(*bufio.Writer).Flush()
	return nil
}

//读取响应头从io流中读取rpc响应头
func (c *clientCodec) ReadResponseHeader(r *rpc.Response) error {
	c.response.ResetHeader()    // 重置clientCodec的响应头部
	data, err := recvFrame(c.r) // 读取请求头字节串
	if err != nil {
		return err
	}
	err = c.response.Unmarshal(data) // 用序列化器继续解码
	if err != nil {
		return err
	}
	c.mutex.Lock()
	r.Seq = c.response.ID              // 填充 r.Seq
	r.Error = c.response.Error         // 填充 r.Error
	r.ServiceMethod = c.pending[r.Seq] // 根据序号填充 r.ServiceMethod
	delete(c.pending, r.Seq)           // 删除pending里的序号
	c.mutex.Unlock()
	return nil
}

//读取响应体从io流中读取rpc响应体
func (c *clientCodec) ReadResponseBody(param interface{}) error {
	if param == nil {
		if c.response.ResponseLen != 0 {
			if err := read(c.r, make([]byte, c.response.ResponseLen)); err != nil {
				return err
			}
		}
		return nil
	}

	respBody := make([]byte, c.response.ResponseLen)
	err := read(c.r, respBody)
	if err != nil {
		return err
	}
	// 校验
	if c.response.Checksum != 0 {
		if crc32.ChecksumIEEE(respBody) != c.response.Checksum {
			return UnexpectedChecksumError
		}
	}

	if c.response.GetCompressType() != c.compressor {
		return CompressorTypeMismatchError
	}
	// 解压
	resp, err := compressor.Compressors[c.response.GetCompressType()].Unzip(respBody)
	if err != nil {
		return err
	}

	return c.serializer.Unmarshal(resp, param)
}

func (c *clientCodec) Close() error {
	return c.c.Close()
}
